package kylin.obs2hive

import kylin.po.JddjInfo
import org.apache.spark.SparkContext
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

/**
  * @ Auther: o2o-rd-0008
  * @ Date:   2020/7/24 10:56
  * @ Param:  ${PARAM}
  * @ Description: 
  */
object JddjObs2Hive {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
//        .master("local[*]")
        .appName(this.getClass.getName)
        .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
        .enableHiveSupport()
        .getOrCreate()

    import spark.implicits._

    var sc: SparkContext = spark.sparkContext
    sc.hadoopConfiguration.set("fs.s3a.access.key", "GAO7EO9FWKPJ8WFCQDME")
    sc.hadoopConfiguration.set("fs.s3a.secret.key", "LZ0xaHBSYKHaJ9ECDbX9f7zin79UZkXfGoNapRPL")
    sc.hadoopConfiguration.set("fs.s3a.endpoint", "https://obs.cn-north-1.myhuaweicloud.com")


    sc.hadoopConfiguration.set("hive.exec.dynamic.partition.mode","nonstrict")

    sc.setLogLevel("ERROR")

    val jddjSourcePath= "s3a://dws-data/split/split_data/2020/jd_daojia/7/"

    val jddjSource: DataFrame = spark.read.orc(jddjSourcePath)

    jddjSource.mapPartitions(partition=>{
      partition.map(lines=>{
        val address: String = lines.getString(0)
        val administrative_region: String = lines.getString(1)
        val aedzId: String = lines.getString(2)
        val brandName: String = lines.getString(3)
        val brandName_cn: String = lines.getString(4)
        val brandName_en: String = lines.getString(5)
        val brandSourceId: String = lines.getString(6)
        val brandValueId: String = lines.getString(7)
        val brand_isLaoZiHao: String = lines.getString(8)
        val brand_state: String = lines.getString(9)
        val brand_type: String = lines.getString(10)
        val categoryId: String = lines.getString(11)
        val categoryName: String = lines.getString(12)
        val city: String = lines.getString(13)
        val city_grade: String = lines.getString(14)
        val city_origin: String = lines.getString(15)
        val county: String = lines.getString(16)
        val coupon: String = lines.getString(17)
        val crawl_date: String = lines.getString(18)
        val district: String = lines.getString(19)
        val district_origin: String = lines.getString(20)
        val economic_division: String = lines.getString(21)
        val firstCategoryId: String = lines.getString(22)
        val fourthCategoryId: String = lines.getString(23)
        val goodRatePercentage: String = lines.getString(24)
        val good_id: String = lines.getString(25)
        val if_city: String = lines.getString(26)
        val if_district: String = lines.getString(27)
        val if_state_level_new_areas: String = lines.getString(28)
        val images: String = lines.getString(29)
        val jd_categoryId: String = lines.getString(30)
        val jd_categoryName: String = lines.getString(31)
        val jd_rootCategoryId: String = lines.getString(32)
        val jd_rootCategoryName: String = lines.getString(33)
        val jd_subCategoryId: String = lines.getString(34)
        val jd_subCategoryName: String = lines.getString(35)
        val latitude: String = lines.getString(36)
        val longitude: String = lines.getString(37)
        val name: String = lines.getString(38)
        val platformId: String = lines.getString(39)
        val platformName: String = lines.getString(40)
        val poor_counties: String = lines.getString(41)
        val priceText: Double = lines.getString(42).toDouble
        val province: String = lines.getString(43)
        val quantity: String = lines.getString(44)
        val regional_ID: String = lines.getString(45)
        val registration_institution: String = lines.getString(46)
        val rootCategoryId: String = lines.getString(47)
        val rootCategoryName: String = lines.getString(48)
        val rural_demonstration_counties: String = lines.getString(49)
        val rural_ecommerce: String = lines.getString(50)
        val salesAmount: Double = lines.getDouble(51)
        val secondCategoryId: String = lines.getString(52)
        val sellCount: Integer = lines.getString(53).toInt
        val serviceTimes: String = lines.getString(54)
        val shopId: String = lines.getString(55)
        val shopImages: String = lines.getString(56)
        val shopName: String = lines.getString(57)
        val shopSellCount: String = lines.getString(58)
        val shopType: String = lines.getString(59)
        val shopUrl: String = lines.getString(60)
        val sku_id: String = lines.getString(61)
        val street: String = lines.getString(62)
        val the_belt_and_road_city: String = lines.getString(63)
        val the_belt_and_road_province: String = lines.getString(64)
        val the_yangtze_river_economic_zone_city: String = lines.getString(65)
        val the_yangtze_river_economic_zone_province: String = lines.getString(66)
        val thirdCategoryId: String = lines.getString(67)
        val timeStamp_date: String = lines.getString(68)
        val title: String = lines.getString(69)
        val town: String = lines.getString(70)
        val urban_agglomerations: String = lines.getString(71)
        val venderId: String = lines.getString(72)
        val dt = "2020-07-30"

        JddjInfo(address, administrative_region, aedzId, brandName, brandName_cn, brandName_en, brandSourceId, brandValueId, brand_isLaoZiHao, brand_state, brand_type, categoryId, categoryName, city, city_grade, city_origin, county, coupon, crawl_date, district, district_origin, economic_division, firstCategoryId, fourthCategoryId, goodRatePercentage, good_id, if_city, if_district, if_state_level_new_areas, images, jd_categoryId, jd_categoryName, jd_rootCategoryId, jd_rootCategoryName, jd_subCategoryId, jd_subCategoryName, latitude, longitude, name, platformId, platformName, poor_counties, priceText, province, quantity, regional_ID, registration_institution, rootCategoryId, rootCategoryName, rural_demonstration_counties, rural_ecommerce, salesAmount, secondCategoryId, sellCount, serviceTimes, shopId, shopImages, shopName, shopSellCount, shopType, shopUrl, sku_id, street, the_belt_and_road_city, the_belt_and_road_province, the_yangtze_river_economic_zone_city, the_yangtze_river_economic_zone_province, thirdCategoryId, timeStamp_date, title, town, urban_agglomerations, venderId,dt)
      })
    }).coalesce(8).write.mode(SaveMode.Append).insertInto("jd_daojia.jd_daojia_info")

  }
}
